package com.gjy.kafka.sp.service;

import com.gjy.kafka.sp.config.KafkaConstant;
import com.gjy.kafka.sp.entity.MessageEntity;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

/**
 * Kafka消息生产者服务
 * 负责向Kafka发送各种类型的消息
 *
 * @author gjy
 * @version 1.0
 * @since 2025-09-07 10:40:53
 */
@Service
public class KafkaProducerService {
    private static final Logger log = LoggerFactory.getLogger(KafkaProducerService.class);

    /**
     * Kafka模板类，提供发送消息的各种方法
     */
    @Resource
    private KafkaTemplate<String, MessageEntity> kafkaTemplate;

    /**
     * 事务Kafka模板类，用于发送事务消息
     */
    @Resource(name = "transactionalKafkaTemplate")
    private KafkaTemplate<String, MessageEntity> transactionalKafkaTemplate;

    /**
     * 消息轨迹服务
     */
    @Resource
    private MessageTraceService messageTraceService;

    /**
     * 发送普通消息
     *
     * @param topic   主题名称
     * @param message 消息实体
     */
    public void sendMessage(String topic, MessageEntity message) {
        // 确保消息ID和创建时间不为空
        if (org.apache.commons.lang3.StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }

        // 记录消息发送前的轨迹
        messageTraceService.recordBeforeSend(message, topic);

        log.info("发送Kafka消息，主题：{}，消息ID：{}，业务类型：{}",
                topic, message.getMessageId(), message.getBusinessType());

        // 发送消息
        ListenableFuture<SendResult<String, MessageEntity>> future = kafkaTemplate.send(topic, message.getMessageId(), message);

        // 处理发送结果
        future.addCallback(new ListenableFutureCallback<SendResult<String, MessageEntity>>() {

            @Override
            public void onSuccess(SendResult<String, MessageEntity> result) {
                log.info("Kafka消息发送成功，主题：{}，消息ID：{}，分区：{}，偏移量：{}",
                        topic, message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());

                // 记录消息发送成功的轨迹
                messageTraceService.recordSendSuccess(message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("Kafka消息发送失败，主题：{}，消息ID：{}",
                        topic, message.getMessageId(), ex);

                // 记录消息发送失败的轨迹
                messageTraceService.recordSendFailure(message.getMessageId(), ex.getMessage());
            }
        });
    }

    /**
     * 发送分区消息
     *
     * @param message 消息实体
     */
    public void sendPartitionMessage(MessageEntity message) {
        // 参数校验
        Objects.requireNonNull(message, "消息实体不能为空");

        // 确保消息ID和创建时间不为空
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }

        String topic = KafkaConstant.PARTITION_TOPIC;

        // 记录消息发送前的轨迹
        messageTraceService.recordBeforeSend(message, topic);

        log.info("发送Kafka分区消息，主题：{}，消息ID：{}，业务ID：{}，业务类型：{}",
                topic, message.getMessageId(), message.getBusinessId(), message.getBusinessType());

        // 发送消息，使用业务ID作为key，配合自定义分区策略
        ListenableFuture<SendResult<String, MessageEntity>> future = kafkaTemplate.send(topic, message.getBusinessId(), message);

        // 处理发送结果
        future.addCallback(new ListenableFutureCallback<SendResult<String, MessageEntity>>() {
            @Override
            public void onSuccess(SendResult<String, MessageEntity> result) {
                log.info("Kafka分区消息发送成功，主题：{}，消息ID：{}，业务ID：{}，分区：{}，偏移量：{}",
                        topic, message.getMessageId(), message.getBusinessId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());

                // 记录消息发送成功的轨迹
                messageTraceService.recordSendSuccess(message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("Kafka分区消息发送失败，主题：{}，消息ID：{}，业务ID：{}",
                        topic, message.getMessageId(), message.getBusinessId(), ex);

                // 记录消息发送失败的轨迹
                messageTraceService.recordSendFailure(message.getMessageId(), ex.getMessage());
            }
        });
    }

    /**
     * 发送事务消息
     *
     * @param message 消息实体
     */
    @Transactional(rollbackFor = Exception.class)
    public void sendTransactionalMessage(MessageEntity message) {
        // 参数校验
        Objects.requireNonNull(message, "消息实体不能为空");

        // 确保消息ID和创建时间不为空
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }

        String topic = KafkaConstant.TRANSACTIONAL_TOPIC;

        // 记录消息发送前的轨迹
        messageTraceService.recordBeforeSend(message, topic);

        log.info("发送Kafka事务消息，主题：{}，消息ID：{}，业务类型：{}",
                topic, message.getMessageId(), message.getBusinessType());

        // 开始事务
        transactionalKafkaTemplate.executeInTransaction(kafkaOperations -> {
            // 发送消息
            SendResult<String, MessageEntity> result;
            try {
                result = kafkaOperations.send(topic, message.getMessageId(), message).get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }

            log.info("Kafka事务消息发送成功，主题：{}，消息ID：{}，分区：{}，偏移量：{}",
                    topic, message.getMessageId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());

            // 记录消息发送成功的轨迹
            messageTraceService.recordSendSuccess(message.getMessageId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());

            // 这里可以添加数据库操作等其他事务操作

            return result;
        });
    }

    /**
     * 创建消息实体
     *
     * @param content      消息内容
     * @param businessType 业务类型
     * @param businessId   业务ID
     * @param extra        额外信息
     * @return 消息实体
     */
    public MessageEntity createMessageEntity(String content, String businessType, String businessId, String extra) {
        MessageEntity message = new MessageEntity();
        message.setMessageId(UUID.randomUUID().toString());
        message.setContent(content);
        message.setBusinessType(businessType);
        message.setBusinessId(businessId);
        message.setCreateTime(LocalDateTime.now());
        message.setExtra(extra);
        return message;
    }

}
